package com.bingxu.flink.action;

import com.bingxu.flink.bean.UserBehavior;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/13 8:24
 * @modifier :
 */

public class PV_Count2 {
    public static void main(String[] args) throws Exception {
        // 运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // load data and transform Java Bean object
        DataStreamSource<String> sourceDS = env.readTextFile("D:\\Source\\JavaProject\\flinkdemo\\input\\UserBehavior.csv");
        SingleOutputStreamOperator<UserBehavior> userBehaviorDS = sourceDS.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String s) throws Exception {
                String[] fields = s.split(",");
                return new UserBehavior(Long.parseLong(fields[0]),
                        Long.parseLong(fields[1]),
                        Integer.parseInt(fields[2]),
                        fields[3],
                        Long.parseLong(fields[4])
                );
            }
        });

        // process 算子处理
        userBehaviorDS
                .filter(element->"pv".equals(element.getBehavior()))
                .process(new ProcessFunction<UserBehavior, Integer>() {
                    private int sum = 0;

                    @Override
                    public void processElement(UserBehavior value, Context ctx, Collector<Integer> out) throws Exception {
                        sum++;
                        out.collect(sum);
                    }
                }).setParallelism(1)
                .print();


        // 执行
        env.execute();
    }
}
